﻿#include "RpcClient.h"
#include "XyNet/Channel/RpcSocketChannel.h"
#include "XyNet/Common/Cast.h"
#include "XyNet/Common/CrossObjectHolder.h"
#include "XyNet/Common/Exception.h"
#include "XyNet/Common/Logger.h"
#include "XyNet/Common/Random.h"
#include "XyNet/Service/RpcServiceManager.h"
#include <boost/algorithm/string.hpp>

namespace XyNet
{
	RpcClient::Config::Config(const string& baseFileName) : RpcEndpoint::Config(baseFileName) {}

	//////////////////////////////////////////////////////////////////////////

	RpcClient::RpcClient(shared_ptr<Config> config)
		: RpcEndpoint("RpcClient", config), _connecting(false), _numberOfTries(0)
	{
		_resolver = make_shared<boost::asio::ip::tcp::resolver>(*getIoService());
		setQueries(getConfig()->getQueries());
		// 虽然在构造函数调用虚函数createSocketChannel，但不影响使用
		_socketChannel = createSocketChannel(
			make_shared<boost::asio::ip::tcp::socket>(*getIoService()));

		getServiceManager()->addService(make_shared<ClientBaseServiceImpl>(this));

		startConnect();
		run(1);
	}

	RpcClient::~RpcClient()
	{
		stop();
	}

	shared_ptr<const RpcClient::Config> RpcClient::getConfig() const
	{
		return XYNET_POINTER_DOWNCAST<const RpcClient::Config>(RpcEndpoint::getConfig());
	}

	shared_ptr<RpcChannel> RpcClient::getChannel()
	{
		return _socketChannel;
	}

	bool RpcClient::isConnectComplete()
	{
		return _socketChannel->isConnectComplete();
	}

	void RpcClient::startConnect()
	{
		auto socket = _socketChannel->getSocket();
		if (_connecting || socket->is_open())
			return;
		_connecting = true;
		++_numberOfTries;

		// 随机选择一个连接
		int queryIndex = Random::getDefault()->nextInt(0, numeric_cast<int>(_queries.size() - 1));
		boost::asio::ip::tcp::resolver::iterator iterator =
			_resolver->resolve(_queries.at(queryIndex));

		// 连接
		boost::asio::async_connect(*socket, iterator,
			bind(&RpcClient::handleConnect, this, boost::asio::placeholders::error));
	}

	void RpcClient::disconnect()
	{
		_socketChannel->stop();
	}

	void RpcClient::setServerErrorHandler(const ServerErrorHandler& handler)
	{
		_serverErrorHandler = handler;
	}

	void RpcClient::handleServerError(shared_ptr<const RpcError> error)
	{
		if (_serverErrorHandler)
			_serverErrorHandler(error);
		else
			Logger::getDefault()->logError(error->getInfo());
	}

	void RpcClient::onReconnect(const boost::system::error_code& error)
	{
		if (!error)
			startConnect();
	}

	void RpcClient::handleConnect(const boost::system::error_code& error)
	{
		auto socket = _socketChannel->getSocket();
		_connecting = false;

		if (error)
		{
			handleSocketException(socket, error, _socketChannel);

			if (_numberOfTries >= getConfig()->getMaxNumberOfTries())
			{
				_numberOfTries = 0;	// 连接失败，清除尝试次数
				throw XYNET_EXCEPTION("连接失败。");
			}
			else
			{
				_reconnectTimer = make_shared<boost::asio::deadline_timer>(*getIoService());
				_reconnectTimer->expires_from_now(boost::posix_time::seconds(
					getConfig()->getTryInterval()));
				_reconnectTimer->async_wait(bind(&RpcClient::onReconnect,
					this, boost::asio::placeholders::error));
			}
		}
		else
		{
			auto localEndpoint = socket->local_endpoint();
			auto remoteEndpoint = socket->remote_endpoint();
			boost::format format("Local %s:%d connect to %s:%d.");
			format % localEndpoint.address() % localEndpoint.port() %
				remoteEndpoint.address() % remoteEndpoint.port();
			Logger::getDefault()->log(format.str());

			_numberOfTries = 0;	// 连接成功，清除尝试次数
			_socketChannel->start();

			// 简化处理，避免多次重连后心跳计时器的事件叠加
			if (!_heartBeatTimer)
			{
				_heartBeatTimer = make_unique<boost::asio::deadline_timer>(*getIoService());
				startHeartBeatTimer();
			}
		}
	}

	void RpcClient::setQueries(const string& queries)
	{
		Logger::getDefault()->log("Queries: " + queries);
		vector<string> queryList;
		boost::split(queryList, queries, boost::is_any_of(";"), boost::token_compress_on);

		// 分割不同的终端
		_queries.clear();
		for (const string& query : queryList)
		{
			// 分割地址和端口
			vector<string> hostAndPort;
			boost::split(hostAndPort, query, boost::is_any_of(":"));
			if (hostAndPort.size() < 2)
				break;
			_queries.emplace_back(hostAndPort.at(0), hostAndPort.at(1));
		}
		assert(!_queries.empty());
	}

	shared_ptr<RpcBroadcastChannel> RpcClient::createBroadcastChannel()
	{
		throw XYNET_EXCEPTION("客户端不允许创建广播通道。");
		return nullptr;
	}

	shared_ptr<RpcSocketChannel> RpcClient::findSocketChannel(const Id& id) const
	{
		if (_socketChannel->getId() == id)
			return _socketChannel;
		return nullptr;
	}

	void RpcClient::handleSocketChannelException(
		shared_ptr<RpcSocketChannel> socketChannel, int errorCode)
	{
		RpcEndpoint::handleSocketChannelException(socketChannel, errorCode);

		if (errorCode != 0 && socketChannel->getSocket()->is_open())
		{
			// 告诉服务端错误（如果可能的话）
			BaseService::SendErrorMethod method(socketChannel);
			method.getRequest()->set_error_code(errorCode);
			method.call([socketChannel](BaseService::SendErrorMethod::Argument* argument)
			{
				socketChannel->stop();
			}, [socketChannel](BaseService::SendErrorMethod::Argument* argument)
			{
				socketChannel->stop();
			});
		}
		else
		{
			socketChannel->stop();
		}
		auto error = RpcErrors::createError(errorCode);
		Logger::getDefault()->logWarning(error->getInfo());
	}

	void RpcClient::startHeartBeatTimer()
	{
		_heartBeatTimer->expires_from_now(
			boost::posix_time::seconds(getConfig()->getHeartBeatInterval()));
		_heartBeatTimer->async_wait(bind(&RpcClient::handleHeartBeat,
			_socketChannel, boost::asio::placeholders::error));
	}

	void RpcClient::handleHeartBeat(weak_ptr<RpcSocketChannel> weakSocketChannel,
		const boost::system::error_code& error)
	{
		if (!error)
		{
			auto socketChannel = weakSocketChannel.lock();
			if (socketChannel)
			{
				if (socketChannel->isConnectComplete())
				{
					BaseService::SendHeartBeatMethod method(socketChannel);
					method.call();
				}
				XYNET_DOWNCAST<RpcClient*>(socketChannel->getEndpoint())->startHeartBeatTimer();
			}
		}
	}

	//////////////////////////////////////////////////////////////////////////

	ClientBaseServiceImpl::ClientBaseServiceImpl(RpcClient* client)
		: _client(client) {}

	void ClientBaseServiceImpl::onSendError(SendErrorMethod::Argument* argument)
	{
		auto errorCode = argument->getRequest()->error_code();
		auto error = argument->getController()->createError(errorCode);
		error->setDetail(argument->getRequest()->error_detail());
		_client->handleServerError(error);
	}

	void ClientBaseServiceImpl::onSendHeartBeat(SendHeartBeatMethod::Argument* argument)
	{
		throw std::logic_error("The method or operation is not implemented.");
	}

	void ClientBaseServiceImpl::onGetUtcTime(GetUtcTimeMethod::Argument* argument)
	{
		throw std::logic_error("The method or operation is not implemented.");
	}

	void ClientBaseServiceImpl::onGetBrokerQueries(GetBrokerQueriesMethod::Argument* argument)
	{
		throw std::logic_error("The method or operation is not implemented.");
	}

}

//////////////////////////////////////////////////////////////////////////

namespace XyNet
{
	XYNET_IMPLEMENT_BASE_SINGLETON(CrossObjectHolder<RpcClient>);

	XYNET_API RpcClient* RpcClientCreate(string* queries, bool crypto)
	{
		try
		{
			auto config = make_shared<RpcClient::Config>();
			config->setAutoHandleMessage(false);	// 导出后只能用单线程
			config->setQueries(*queries);

			return CrossObjectHolder<RpcClient>::getInstance()->add(
				make_shared<RpcClient>(config));
		}
		catch (const exception& e)
		{
			HandleException(e);
			return nullptr;
		}
	}

	XYNET_API void RpcClientDestroy(RpcClient* client)
	{
		try
		{
			CrossObjectHolder<RpcClient>::getInstance()->remove(client);
		}
		catch (const exception& e)
		{
			HandleException(e);
		}
	}

	XYNET_API RpcChannel* RpcClientGetSocketChannel(RpcClient* client)
	{
		try
		{
			return client->getChannel().get();
		}
		catch (const exception& e)
		{
			HandleException(e);
			return nullptr;
		}
	}

	XYNET_API bool RpcClientIsConnectComplete(RpcClient* client)
	{
		try
		{
			return client->isConnectComplete();
		}
		catch (const exception& e)
		{
			HandleException(e);
			return false;
		}
	}

	XYNET_API void RpcClientStartConnect(RpcClient* client)
	{
		try
		{
			return client->startConnect();
		}
		catch (const exception& e)
		{
			HandleException(e);
		}
	}
}